package org.example.mall.admin.common.es;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.example.mall.admin.configuration.exception.BaseException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.io.IOException;
import java.util.function.Consumer;

/**
 * es 高级客户端
 *
 * @author hzq
 * @date 2022/1/14 21:17
 */
public interface RestHighLevelClientService {

    /**
     * Changed password for user apm_system
     * PASSWORD apm_system = vKMlnTcGco57jfOiR1ae
     *
     * Changed password for user kibana_system
     * PASSWORD kibana_system = f1OeU34vbZUWiF5bB3ye
     *
     * Changed password for user kibana
     * PASSWORD kibana = f1OeU34vbZUWiF5bB3ye
     *
     * Changed password for user logstash_system
     * PASSWORD logstash_system = J0P3X0MhtOvRU594knFv
     *
     * Changed password for user beats_system
     * PASSWORD beats_system = rgDCIYajbWwizTgFU6gl
     *
     * Changed password for user remote_monitoring_user
     * PASSWORD remote_monitoring_user = r5dzWkNmOuxybzAxHaey
     *
     * Changed password for user elastic
     * PASSWORD elastic = B4XjMTWUG7Y8vXwh0HSm
     */

    /**
     * 日志
     */
    Logger logger = LoggerFactory.getLogger(RestHighLevelClientService.class);

    /**
     * @return Rest高级客户端
     */
    RestHighLevelClient restHighLevelClient();

    /**
     * @return 索引名
     */
    String index();

    /**
     * @return ObjectMapper
     */
    ObjectMapper objectMapper();

    /**
     * @return ThreadPoolTaskExecutor
     */
    ThreadPoolTaskExecutor applicationTaskExecutor();

    /**
     * @return 同步到ES状态
     */
    Consumer<String> syncEsStatus();

    /**
     * 创建索引
     */
    default void createIndex() {
        try {
            CreateIndexRequest request = new CreateIndexRequest(index());
            CreateIndexResponse response = restHighLevelClient().indices().create(request, RequestOptions.DEFAULT);
            logger.info(" whether all of the nodes have acknowledged the request : {}", response.isAcknowledged());
            logger.info(" Indicates whether the requisite number of shard copies were started for each shard in the index before timing out :{}", response.isShardsAcknowledged());
        } catch (IOException e) {
            logger.error(e.getLocalizedMessage());
            throw new BaseException("创建索引 {" + index() + "} 失败");
        }
    }

    /**
     * 删除索引
     */
    default void deleteIndex() {
        try {
            DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(index());
            restHighLevelClient().indices().delete(deleteIndexRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error(e.getLocalizedMessage());
            throw new BaseException("删除索引 {" + index() + "} 失败");
        }
    }

    /**
     * @return 是否存在索引
     */
    default boolean existsIndex() {
        try {
            GetIndexRequest getIndexRequest = new GetIndexRequest(index());
            return restHighLevelClient().indices().exists(getIndexRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error(e.getLocalizedMessage());
            throw new BaseException("检测索引 {" + index() + "} 失败");
        }
    }

    /**
     * 保存数据
     *
     * @param id    id
     * @param obj   数据
     * @param clazz 数据类型
     */
    default void asyncSaveEs(String id, Object obj, Class<?> clazz) {
        applicationTaskExecutor().execute(() -> {
            try {
                String strJson;
                if (obj.getClass().isAssignableFrom(clazz)) {
                    strJson = objectMapper().writeValueAsString(obj);
                }
                // https://blog.csdn.net/hellozhxy/article/details/90938381
                // 只保存指定属性 减少ES存储量 fileSystem cache
                else {
                    Object target = clazz.newInstance();
                    BeanUtils.copyProperties(obj, target);
                    strJson = objectMapper().writeValueAsString(target);
                }
                IndexRequest request = new IndexRequest(index()).id(id).source(strJson, XContentType.JSON);
                restHighLevelClient().index(request, RequestOptions.DEFAULT);
                syncEsStatus().accept(id);
            } catch (IOException | InstantiationException | IllegalAccessException e) {
                logger.error("保存索引: {} \n数据: {} \n失败消息: {}", index(), obj, e.getLocalizedMessage());
            }
        });
    }

    /**
     * 修改
     *
     * @param id    id
     * @param obj   数据
     * @param clazz 数据类型
     */
    default void asyncUpdateEs(String id, Object obj, Class<?> clazz) {
        applicationTaskExecutor().execute(() -> {
            try {
                String strJson;
                if (obj.getClass().isAssignableFrom(clazz)) {
                    strJson = objectMapper().writeValueAsString(obj);
                } else {
                    Object target = clazz.newInstance();
                    BeanUtils.copyProperties(obj, target);
                    strJson = objectMapper().writeValueAsString(target);
                }
                UpdateRequest updateRequest = new UpdateRequest(index(), id).doc(strJson, XContentType.JSON);
                restHighLevelClient().update(updateRequest, RequestOptions.DEFAULT);
                syncEsStatus().accept(id);
            } catch (IOException | InstantiationException | IllegalAccessException e) {
                logger.error("更新索引: {} \n数据: {} \n失败消息: {}", index(), obj, e.getLocalizedMessage());
            }
        });
    }

    /**
     * 删除
     *
     * @param id id
     */
    default void asyncDeleteEs(String id) {
        applicationTaskExecutor().execute(() -> {
            try {
                DeleteRequest deleteRequest = new DeleteRequest(index(), id);
                restHighLevelClient().delete(deleteRequest, RequestOptions.DEFAULT);
                syncEsStatus().accept(id);
            } catch (IOException e) {
                logger.error("删除索引: {} \n数据id: {} \n失败消息: {}", index(), id, e.getLocalizedMessage());
            }
        });
    }

    /**
     * 查询
     */
    default SearchResponse search(SearchSourceBuilder searchSourceBuilder) {
        return search(searchSourceBuilder, index());
    }

    /**
     * 查询
     */
    default SearchResponse search(SearchSourceBuilder searchSourceBuilder, String... index) {
        try {
            SearchRequest searchRequest = new SearchRequest(index, searchSourceBuilder);
            return restHighLevelClient().search(searchRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error(e.getLocalizedMessage());
            e.printStackTrace();
        }
        return null;
    }

}
